package com.bank.rabbitmq;

import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeoutException;

/**
 * 测试类
 * 了解rabbitmq的底层java客户端
 */
public class RabbitMqDemo01 {
    /*
    使用java语言 无论链接 mysql还是rabbitmq都需要客户端连接对象
     */
    private Channel channel;//连接通道 有了这个通道 可以操作rabbitmq
    //测试框架 注解
    @Before
    public void init() throws IOException, TimeoutException {
        //长连接 连接工厂 工厂是一种代码设计模式 : 构造一些复杂对象 使用工厂现成的方法 减低使用者使用成本
        ConnectionFactory factory = new ConnectionFactory();
        //需要创建rabbitmq连接的参数 ip port username password
        factory.setHost("localhost");
        factory.setPort(5672);
        factory.setUsername("guest");
        factory.setPassword("guest");
        Connection connection = factory.newConnection();
        channel = connection.createChannel();
    }

    //声明组件 入门声明一个队列 queue01
    @Test
    public void declareQueue() throws IOException {
        Map<String,Object> args=new HashMap<>();
        //队列最大未使用存活时间
        //args.put("x-expires",20000);
        channel.queueDeclare("queue01",false,false,false,args);
    }

    //生产发送消息
    @Test
    public void sendMessage() throws IOException {
        //发送的消息组成
        //1. 消息体  二进制 需要将发送的数据做序列化 比如字符串 byte[]
        //2. 消息携带 路由key值(大部分路由key可以决定消息去向) 交换机有一个消息转发的过程
        //3. 消息头 属性 表示消息的一些特点 比如 appId userId messageId...
        String msg="hello world";
        byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
        //发送
        channel.basicPublish("","queue01",null,bytes);
    }

    @Test
    public void consume() throws IOException {
        //1.调用消费方法
        //2.拿到消息后的处理逻辑true
        channel.basicConsume("queue01",false,
                (consumerTag,message)->{
                    //handle 当消息达到消费者后 执行消费逻辑 consumerTag表示消费者id,message传递消息数据
                    System.out.println("consumerTag="+consumerTag);
                    //消息包含 来源 交换机 消息属性 消息体
                    byte[] body = message.getBody();
                    //反序列化 将二进制解析成内存数据
                    String msg=new String(body);
                    System.out.println("msg="+msg);
                    //如果消息处理成功 确认成功
                    //channel.basicAck(message.getEnvelope().getDeliveryTag(),false);
                    //如果消息处理失败 确认拒绝 重新入队 可以决定 是否允许消息重新入队
                    channel.basicReject(message.getEnvelope().getDeliveryTag(),true);
                },
                consumerTag -> {
                    //handle 消费者取消 断开消费的时候调用 有时候需要这个方法为消费者释放资源
                }
        );
        //当前线程执行速度太快 异步处理过程没有展示
        while(true);
    }

}
